package mr

import (
	"fmt"
	"hash/fnv"
	"net/rpc"
	"os"
	"time"
)

// Map functions return a slice of KeyValue.
type KeyValue struct {
	Key   string
	Value string
}

// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
// 该函数能够根据Map中返回的Key值决定该KeyValue的Reduce任务编号
// 注意 这里只是生成了一个Reduce任务编号 但是并没有被限制在nReduce中
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

// main/mrworker.go calls this function.
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	// Your worker implementation here.
	Init_WorkerLogger()         // 初始化Worker节点日志
	time.Sleep(time.Second * 2) // 等待Master初始化

	// 工作循环
	for {
		MrLoger.Print("Worker One Loop Start")
		// if isDone() {
		// 	MrLoger.Print("Master is Done")
		// 	return
		// }
		// 获取任务
		resp, ok := RequestTask()
		if !ok {
			MrLoger.Print("RequestTask() Error Once")
			resp, ok = RequestTask()
			if !ok {
				MrLoger.Print("RequestTask() Error*2 and Stop")
				return
			}
		}

		// 处理请求
		if resp.Tops == OPS_WAIT || resp.Tops == OPS_NONE {
			// 启动休眠
			MrLoger.Printf("Worker WAIT then Retry")
			time.Sleep(time.Second)
			continue
		} else if resp.Tops == OPS_STOP {
			MrLoger.Printf("Worker STOP and No More Task")
			return
		}

		// // 50%概率等待12秒
		// t := time.Now().UnixMicro()
		// if t%2 == 0 {
		// 	MrLoger.Printf("Worker Random Sleep")
		// 	time.Sleep(time.Second * 12)
		// }

		var resTask *Task = nil
		nReduce := resp.ReduceCnt
		task := resp.WorkTask
		if resp.Tops == OPS_M { //Map
			resTask = Worker_Map(&task, nReduce, mapf)
		} else if resp.Tops == OPS_R { //Reduce
			resTask = Worker_Reduce(&task, nReduce, reducef)
		}

		if resTask == nil {
			// 没有运行结果
			continue
		}

		// 循环汇报 - Report返回的OPS_WAIT表明等待后再汇报
		// 三次汇报出错则不再执行
		reportCnt := 0
		for {
			if reportCnt == 3 {
				MrLoger.Print("Worker Report-Error*3 goto Next Loop")
				break
			}
			reportCnt++

			// 汇报
			MrLoger.Printf("Worker ReportTask Cnt=%v", reportCnt)
			resT, ok := ReportTask(resTask)
			if !ok {
				resT, ok = ReportTask(resTask)
				if !ok {
					return
				}
			}

			taskInfo := fmt.Sprintf("TaskID:%v,TaskRunID:%v", resTask.TaskID, resTask.TaskRunID)
			if resT.Tops == OPS_WAIT {
				MrLoger.Printf("Worker ReportTask Sleep %v", taskInfo)
				time.Sleep(time.Second)
				continue // 继续尝试汇报
			} else if resT.Tops == OPS_STOP {
				MrLoger.Printf("Worker ReportTask Stop %v", taskInfo)
				return // 要求系统停止
			} else if resT.Tops == OPS_NONE {
				MrLoger.Printf("Worker ReportTask None %v", taskInfo)
				break // 汇报成功
			}
		} // report for
	} // main for
}

// // 汇报任务
// func isDone() bool {
// 	// 请求参数
// 	stat := false
// 	// RPC调用
// 	res := call("Master.Done", nil, &stat)
// 	// 请求出错
// 	if !res {
// 		return false
// 	}
// 	return stat
// }

// 汇报任务
func ReportTask(resTask *Task) (*ReportRespone, bool) {
	// 请求参数
	req := ReportRequest{
		Tops:     OPS_REQUEST,
		WorkTask: *resTask,
	}
	resp := ReportRespone{}
	// RPC调用
	res := call("Master.ReportTask", &req, &resp)
	// res := call("ReportTask", &req, &resp)
	// 请求出错
	if !res {
		return nil, false
	}
	return &resp, true
}

// 请求任务
func RequestTask() (*AssignRespone, bool) {
	// 请求参数
	req := AssignRequest{
		Tops: OPS_REQUEST,
	}
	resp := AssignRespone{}
	// RPC调用
	res := call("Master.AssignTask", &req, &resp)
	// res := call("AssignTask", &req, &resp)
	// 请求出错
	if !res {
		return nil, false
	}
	return &resp, true
}

// 键值对结果写入文件
func MapResult2File(taskRunID int, kvlis []KeyValue, nReduce int) ([]TagFile, error) {
	tagMap := make(map[string]bool)
	tagFiles := make([]TagFile, 0)
	mfp := GenMidFilePool()
	for _, kv := range kvlis {
		ytag := ihash(kv.Key) % nReduce
		filename := fmt.Sprintf("%v-M-%v.json", taskRunID, ytag)
		// MrLoger.Printf("kvlis MF: %v", filename)
		mfp.Add(filename, &kv)
		// 注册TagFile
		if _, ok := tagMap[filename]; !ok {
			tagFiles = append(tagFiles, GenTagFile(filename))
			tagMap[filename] = true
		}
	}
	// 查看tagMap的所有文件名
	keys := make([]string, 0, len(tagMap))
	for k := range tagMap {
		keys = append(keys, k)
	}
	MrLoger.Printf("tagMap's Key: %v", keys)
	// 文件写入磁盘
	err := mfp.PoolWrite()
	if err != nil {
		return nil, err
	}
	return tagFiles, nil
}

// Map工作
func Worker_Map(task *Task, nReduce int,
	mapf func(string, string) []KeyValue) *Task {
	MrLoger.Printf("Worker-Map TaskID:%v,TaskRunID:%v", task.TaskID, task.TaskRunID)
	files := task.TaskFile
	kvlis := make([]KeyValue, 0)
	// 逐个文件处理
	for _, tf := range files {
		// 读取文件
		filename := tf.GetFileName()
		MrLoger.Printf("Worker-Map filename:%v", filename)
		f, err := os.Open(filename)
		if err != nil {
			MrLoger.Printf("err: %v\n", err)
			return nil
		}
		defer f.Close()
		// 读取文件内容
		in, err := Read2String(f)
		if err != nil {
			MrLoger.Printf("err: %v\n", err)
			return nil
		}
		// 调用Map函数
		kvlis = append(kvlis, mapf(filename, in)...)
	}
	// MrLoger.Printf("kvlis: %v\n", kvlis)
	// 返回结果任务的描述文件
	outFiles, err := MapResult2File(task.TaskRunID, kvlis, nReduce)
	if err != nil {
		fmt.Printf("err: %v\n", err)
		return nil
	}
	res := Task{
		TaskID:    task.TaskID,
		TaskRunID: task.TaskRunID,
		TaskType:  task.TaskType,
		TaskFile:  outFiles,
	}
	return &res
}

// Reduce工作
func Worker_Reduce(task *Task, nReduce int,
	reducef func(string, []string) string) *Task {
	MrLoger.Printf("Worker-Reduce TaskID:%v,TaskRunID:%v", task.TaskID, task.TaskRunID)

	// 结果文件
	filename := fmt.Sprintf("%v-R-%v.json", task.TaskRunID, task.TaskID-1)
	cacheFile := fmt.Sprintf("%v-C-%v.json", task.TaskRunID, task.TaskID-1)
	mfC := GenMidFile(cacheFile)
	mfile := GenMidFile(filename)

	// 汇总不同的M文件的输出
	for _, tf := range task.TaskFile {
		MrLoger.Printf("Worker-Reduce filename:%v", tf.FileName)
		mf, err := ReadMidFile(tf.FileName)
		if err != nil {
			MrLoger.Printf("err: %v\n", err)
			return nil
		}
		for _, kvl := range mf.KVsMap {
			for _, valx := range kvl.ValueList {
				mfC.AddKv(&KeyValue{
					Key:   kvl.Key,
					Value: valx,
				})
			}
		}
	}

	// 调用Reduce
	for k, kvl := range mfC.KVsMap {
		Rout := reducef(k, kvl.ValueList)
		mfile.AddKv(&KeyValue{
			Key:   kvl.Key,
			Value: Rout,
		})
	}

	// 结果文件写入磁盘
	err := WriteMidFile(&mfile)
	if err != nil {
		MrLoger.Printf("err: %v\n", err)
		return nil
	}

	// 返回的Task
	res := Task{
		TaskID:    task.TaskID,
		TaskRunID: task.TaskRunID,
		TaskType:  task.TaskType,
		TaskFile:  []TagFile{GenTagFile(filename)},
	}

	return &res
}

// example function to show how to make an RPC call to the master.
//
// the RPC argument and reply types are defined in rpc.go.
func CallExample() {

	// declare an argument structure.
	args := ExampleArgs{}

	// fill in the argument(s).
	args.X = 99

	// declare a reply structure.
	reply := ExampleReply{}

	// send the RPC request, wait for the reply.
	call("Master.Example", &args, &reply)

	// reply.Y should be 100.
	fmt.Printf("reply.Y %v\n", reply.Y)
}

// send an RPC request to the master, wait for the response.
// usually returns true.
// returns false if something goes wrong.
// worker线程发起RPC调用的函数
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := masterSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		MrLoger.Printf("dialing:%v", err)
		return false
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	MrLoger.Print(err)
	return false
}
